#include<MQTTAsync.h>
#include<stdio.h>
#include<unistd.h>
#include <string.h>
#include <time.h>
#include <mysql/mysql.h>


#define SERVER "114.215.139.160"
#define CID "ljh"
#define PAYLOAD "hello!"
#define TOPIC "run"
#define TOPIC2 "tem"
#define TOPIC3 "hum"

MQTTAsync client=NULL;

int ret;

int subscribed =0;
int finished =0;


time_t timep;
char str[60];
MYSQL com_mysql; //声明一个数据库连接句柄
char sqlcommand[100];
MYSQL_RES *pRes;
MYSQL_ROW hs;

void connetmysql(void)
{
	if (mysql_init(&com_mysql) == NULL)
	{
		printf("Cannot init mysql!\n");
	}
	if (mysql_real_connect(&com_mysql, "www.ljhong.icu", "LJH", "ljhiot", "LJH", 3306, NULL, 0) == NULL)
	{
		printf("%s\n", mysql_error(&com_mysql));
	}
}

void writehum(char humdata[])
{
	time(&timep);
	struct tm *n = localtime(&timep);
	sprintf(str, "%d/%d/%d %d:%d:%d", n->tm_year + 1900, n->tm_mon + 1, n->tm_mday, n->tm_hour, n->tm_min, n->tm_sec);

	sprintf(sqlcommand, "insert into hum(hum,time)values('%s','%s')", humdata, str);
	printf("%s\n", sqlcommand);
	if (mysql_query(&com_mysql, sqlcommand) != 0)
	{
		printf("%s\n", mysql_error(&com_mysql));
	}
	else
	{
		printf("succeed\n");
	}
}

void writetem(char temdata[])
{
	time(&timep);
	struct tm *n = localtime(&timep);
	sprintf(str, "%d/%d/%d %d:%d:%d", n->tm_year + 1900, n->tm_mon + 1, n->tm_mday, n->tm_hour, n->tm_min, n->tm_sec);

	sprintf(sqlcommand, "insert into tem(tem,time)values('%s','%s')", temdata, str);
	printf("%s\n", sqlcommand);
	if (mysql_query(&com_mysql, sqlcommand) != 0)
	{
		printf("%s\n", mysql_error(&com_mysql));
	}
	else
	{
		printf("succeed\n");
	}
}

void querymsql(int flg)
{
	switch (flg)
	{
	case 1:
		sprintf(sqlcommand, "select * from hum");
		if (mysql_query(&com_mysql, sqlcommand) != 0)
		{
			printf("%s\n", mysql_error(&com_mysql));
			break;
		}
		if ((pRes = mysql_store_result(&com_mysql)) == NULL)
		{
			printf("%s\n", mysql_error(&com_mysql));
			break;
		}
		printf("TABLE:[ hum ]");
		break;
	case 2:
		sprintf(sqlcommand, "select * from tem");
		if (mysql_query(&com_mysql, sqlcommand) != 0)
		{
			printf("%s\n", mysql_error(&com_mysql));
			break;
		}
		if ((pRes = mysql_store_result(&com_mysql)) == NULL)
		{
			printf("%s\n", mysql_error(&com_mysql));
			break;
		}
		printf("TABLE:[ tem ]");	
		break;
	}

	//show result
	printf("===========================\n");
	printf("[  id    ][   data   ][  time  ]\n");
	printf("-----------------------------\n");
	while (hs = mysql_fetch_row(pRes))
	{
		/* code */
		printf("[  %-5s][  %-6s][%s]\n", hs[0], hs[1], hs[2]);
	}
	mysql_free_result(pRes);
}

// --

void onSubscribe(void* context, MQTTAsync_successData* response)
{
	printf("Subscribe succeeded\n");
	subscribed = 1;
}
void onSubscribeFailure(void* context, MQTTAsync_failureData* response)
{
	printf("Subscribe failed, rc %d\n", response->code);
	finished = 1;
}

void connlost(void *context, char *cause)
{

	printf("connect lost!!\n");
	finished = 1;
	
}
//消息回调
int msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
	char data[3];
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: %.*s\n", message->payloadlen, (char*)message->payload);
	sprintf(data,"%s",topicName);
	printf("table:%s\n",data);

	if(strcmp(TOPIC2,data) == 0){
		writetem((char*)message->payload);
	}
	if(strcmp(TOPIC3,data) == 0){
		writehum((char*)message->payload);
	}

    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);
    return 1;
}

// void onConnect(void* context, MQTTAsync_successData* response)
void onConnect(void* context, MQTTAsync_successData* response){
  	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	int rc;

	printf("Successful connection\n");

	printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n", TOPIC2, CID, 1);
	opts.onSuccess = onSubscribe;
	opts.onFailure = onSubscribeFailure;
	opts.context = client;
	//订阅
	if ((rc = MQTTAsync_subscribe(client, TOPIC2, 1, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start subscribe, return code %d\n", rc);
		finished = 1;
	}

	if ((rc = MQTTAsync_subscribe(client, TOPIC3, 1, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start subscribe, return code %d\n", rc);
		finished = 1;
	}

}



void createmqtt(){
   // 声明客户端
    // 创建客户端
    if ((ret = MQTTAsync_create(&client, SERVER, CID,
     MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to create client object, return code %d\n", ret);
	}
	//订阅回调
	if ((ret = MQTTAsync_setCallbacks(client, NULL, connlost, msgarrvd, NULL)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to set callbacks, return code %d\n", ret);
	}
    MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onConnect;
	// conn_opts.onFailure = onConnectFailure;
	conn_opts.context = client;
	// 连接客户端
	if ((ret = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", ret);
	}



}

void sendmqtt(char str[]){

	MQTTAsync_responseOptions opts= MQTTAsync_responseOptions_initializer;
	//opts.onSuccess = onSend;
	//opts.onFailure = onSendFailure;
	opts.context = client;

	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	pubmsg.payload = str;
	pubmsg.payloadlen = (int)strlen(str);
	pubmsg.qos = 1;
	pubmsg.retained = 0;

	if ((ret = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start sendMessage, return code %d\n", ret);
	}else
	{
		printf("success sendmessgae!\n");
	}	
}


int main(){
 
	createmqtt();
	connetmysql();
    
	
	while (1)
    {
        sleep(5);
		sendmqtt("runing");
		
    }
    
    return 0;
}